/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.operation;

import java.io.IOException;
import java.util.Enumeration;
import java.util.Properties;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;

import cgl.granules.dataset.StreamingAccess;
import cgl.granules.exception.CommunicationsException;
import cgl.granules.exception.DeploymentException;
import cgl.granules.exception.MarshallingException;
import cgl.granules.exchange.ConditionFlag;
import cgl.granules.exchange.DeploymentMessage;
import cgl.granules.exchange.Exchange;
import cgl.granules.exchange.ExchangeFactory;
import cgl.granules.exchange.ExchangeType;
import cgl.granules.exchange.ExecutionStatus;
import cgl.granules.exchange.ResourceDiscovery;
import cgl.granules.scheduler.ResourceMetrics;
import cgl.granules.scheduler.StreamingService;
import cgl.granules.scheduler.TaskLifecycleMetrics;
import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.util.UUIDRetriever;

/**
 * This class manages the submission, tracking and cancellation of Jobs and
 * their constituent tasks.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class JobSubmission implements NBEventListener {

    private static final JobSubmission instance = new JobSubmission();
    private StreamingService streamingService;
    private EventConsumer consumer;
    private EventProducer producer;
    private UUIDRetriever retriever;
    private ExchangeFactory exchangeFactory;
    private String trackerBinding;
    private ConcurrentHashMap<String, ProgressTrackerWidget> progressTrackers;
    private String deploymentStream;
    private String discoveryStream = "Granules/Resources/Discovery";
    private String jobberStream;
    private String moduleName = "JobSubmission: ";
    private ConcurrentHashMap<String, ResourceMetrics> resources;
    private Vector<String> resourceOrder;
    private boolean debug = false;


    private JobSubmission() {
        streamingService = StreamingService.getInstance();
        exchangeFactory = ExchangeFactory.getInstance();
        retriever = UUIDRetriever.getInstance();
        trackerBinding = retriever.getRandomBasedUUIDAsString();
        progressTrackers =
            new ConcurrentHashMap<String, ProgressTrackerWidget>();
        resources = new ConcurrentHashMap<String, ResourceMetrics>();
        resourceOrder = new Vector<String>();
    }


    public static JobSubmission getInstance() {
        return instance;
    }


    /**
     * Initializes communications based on the specified parameters.
     * 
     * @param configurationFile
     *            The configuration file
     * @param entityId
     *            The ID for the entity
     * @param props
     *            The properties for streaming
     * @param commType
     *            The communications type
     * @throws CommunicationsException
     *             If there are problems sending messages over the network
     */
    public void initialize(String configurationFile, int entityId,
                           Properties props, String commType)
        throws CommunicationsException {

        streamingService.initialize(configurationFile, entityId, props,
                                    commType);
        consumer = streamingService.createConsumer(this);
        producer = streamingService.createProducer();
        deploymentStream = "Resource/Deployments/Node";
        jobberStream =
            "JobSubmission/" + retriever.getRandomBasedUUIDAsString();
        consumeStream(jobberStream, StreamingAccess.STRING_SYNOPSIS);
    }


    /**
     * Discovers the available computational resources.
     * 
     * @throws IOException
     * @throws MarshallingException
     * @throws CommunicationsException
     * @throws DeploymentException
     */
    public void discoverResource() throws IOException, MarshallingException,
        CommunicationsException, DeploymentException {
        int resourceType = 1;
        ResourceDiscovery resourceRequest =
            exchangeFactory.createResourceDiscoveryRequest(resourceType,
                                                           jobberStream);
        byte[] requestBytes = resourceRequest.getBytes();

        publishToStream(discoveryStream, requestBytes);

        for (int i = 0; i < 6; i++) {
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.print("*");
        }
        System.out.print("\n");

        if (resources.size() == 0) {
            StringBuffer report = new StringBuffer();
            report.append("\nZERO resources were discovered. Please make sure");
            report.append(" that you have started AT LEAST one resource and ");
            report.append("configured it to CONNECT TO the correct BROKER ");
            report.append("(or broker network) that you are connecting to.\n");
            throw new DeploymentException(report.toString());
        }
    }


    /**
     * Retrieves the number of available resources.
     * 
     * @return
     */
    public int getNumberOfAvailableResources() {
        return resources.size();
    }


    /**
     * Creates a Job Identifier
     * 
     * @return The created job identifier.
     */
    public String createJobIdentifier() {
        String jobIdentifier = retriever.getRandomBasedUUIDAsString();
        return jobIdentifier;
    }


    /**
     * Submit the given operation for execution. A given operation can be
     * executed at multiple nodes. The operation returns a JobLifecyle Object
     * that allows an entity to track that status of the JOb's execution at
     * different stages.
     * 
     * @param operation
     *            The operation that needs to be executed at the remote sites
     * @param jobLifecycleObserver
     *            The observer being registered, and on whom the appropriate
     *            methods outlining the executing of the job will be invoked.
     * @return The JobLifecycle object that allows one to track the status of
     *         the individual tasks that comprise the Job
     * @throws CommunicationsException
     *             If there are problems sending messages over the network
     * @throws DeploymentException
     *             if there are problems during the deployment process * @throws
     *             DeploymentException
     * @throws MarshallingException
     *             If there are problems Marshalling the operation in a
     *             deployment request
     */
    public ProgressTracker submit(Operation operation,
                                  JobLifecycleObserver jobLifecycleObserver)
        throws CommunicationsException, DeploymentException,
        MarshallingException {
        String jobIdentifier = retriever.getRandomBasedUUIDAsString();
        return submit(operation, deploymentStream, jobIdentifier,
                      jobLifecycleObserver);
    }


    /**
     * Submit the given operation for execution. The submission request is
     * targeted to a specific resource. A given operation can be executed at
     * multiple nodes. The operation returns a JobLifecyle Object that allows an
     * entity to track that status of the JOb's execution at different stages.
     * 
     * @param operation
     *            The operation that needs to be executed at the remote sites
     * @param resourceDeploymentStream
     *            The deployment stream associated with the resource.
     * @param jobLifecycleObserver
     *            The observer being registered, and on whom the appropriate
     *            methods outlining the executing of the job will be invoked.
     * @return The JobLifecycle object that allows one to track the status of
     *         the individual tasks that comprise the Job
     * @throws CommunicationsException
     *             If there are problems sending messages over the network
     * @throws DeploymentException
     *             if there are problems during the deployment process * @throws
     *             DeploymentException
     * @throws MarshallingException
     *             If there are problems Marshalling the operation in a
     *             deployment request
     */
    public ProgressTracker submit(Operation operation,
                                  String resourceDeploymentStream,
                                  JobLifecycleObserver jobLifecycleObserver)
        throws CommunicationsException, DeploymentException,
        MarshallingException {
        String jobIdentifier = retriever.getRandomBasedUUIDAsString();
        return submit(operation, resourceDeploymentStream, jobIdentifier,
                      jobLifecycleObserver);
    }


    /**
     * This method allows one to group a set of related tasks together in a Job.
     * The specified jobIdentifier should have been previously created using the
     * <code>createJobIdentifier()</code> method.
     * 
     * @param operation
     *            The operation that needs to be executed at the remote sites
     * @param resourceDeploymentStream
     *            The deployment stream associated with the resource.
     * @param jobIdentifier
     *            The identifier of the Job group.
     * @param jobLifecycleObserver
     *            The observer being registered, and on whom the appropriate
     *            methods outlining the executing of the job will be invoked.
     * @return The JobLifecycle object that allows one to track the status of
     *         the individual tasks that comprise the Job
     * @throws CommunicationsException
     *             If there are problems sending messages over the network
     * @throws DeploymentException
     *             if there are problems during the deployment process * @throws
     *             DeploymentException
     * @throws MarshallingException
     *             If there are problems Marshalling the operation in a
     *             deployment request
     */
    public synchronized ProgressTracker submit(
                                               Operation operation,
                                               String resourceDeploymentStream,
                                               String jobIdentifier,
                                               JobLifecycleObserver jobLifecycleObserver)
        throws CommunicationsException, DeploymentException,
        MarshallingException {
        if (jobLifecycleObserver == null) {
            throw new NullPointerException(moduleName + "The specified Job"
                + "LifecycleObserver is null.");
        }
        DeploymentMessage deploymentRequest =
            exchangeFactory.createDepoymentRequest(jobIdentifier, operation,
                                                   jobberStream);
        sendExchangeRequest(deploymentRequest, resourceDeploymentStream);

        ProgressTrackerWidget tracker = progressTrackers.get(jobIdentifier);
        if (tracker == null) {
            tracker =
                new ProgressTrackerWidget(trackerBinding, jobIdentifier,
                                          jobLifecycleObserver);
            progressTrackers.put(jobIdentifier, tracker);
        }
        tracker.incrementNumberOfDeployments();
        return tracker;
    }


    /**
     * Attempts to kill the job while it is in execution at different sites
     * 
     * @param jobIdentifier
     *            The ID of the Job that needs to be killed
     * @throws DeploymentException
     *             If there are problems in locating the job, or if it has
     *             already been completed or canceled
     * @throws MarshallingException
     *             If there are problems marshalling the request
     * @throws CommunicationsException
     *             If there are problems sending the request over the network.
     */
    public void killJob(String jobIdentifier) throws DeploymentException,
        CommunicationsException, MarshallingException {
        ProgressTracker tracker = progressTrackers.get(jobIdentifier);
        Enumeration<String> taskIdentifiers = tracker.getTaskIdentifiers();

        while (taskIdentifiers.hasMoreElements()) {
            String taskIdentifier = taskIdentifiers.nextElement();
            abortTask(jobIdentifier, taskIdentifier);
        }
    }


    /**
     * Send an abort request for a specific task.
     * 
     * @param jobIdentifier
     *            The Job that this task is a part of
     * @param taskIdentifier
     *            The ID of the task that needs to be aborted
     * @throws DeploymentException
     *             If there are problems in locating the job, or if it has
     *             already been completed or canceled
     */
    public void abortTask(String jobIdentifier, String taskIdentifier)
        throws CommunicationsException, DeploymentException,
        MarshallingException {
        ProgressTracker tracker = progressTrackers.get(jobIdentifier);
        String deploymentStream = tracker.getDeploymentStream(taskIdentifier);

        DeploymentMessage deploymentRequest =
            exchangeFactory.createDeploymentStatusOrAbortRequest(
                                                                 jobIdentifier,
                                                                 taskIdentifier,
                                                                 jobberStream,
                                                                 DeploymentMessage.ABORT);
        sendExchangeRequest(deploymentRequest, deploymentStream);
    }


    /**
     * Retrieve the status of a task.
     * 
     * @param jobIdentifier
     * @param taskIdentifier
     * @throws CommunicationsException
     * @throws DeploymentException
     * @throws MarshallingException
     */
    public void getStatusOfTask(String jobIdentifier, String taskIdentifier)
        throws CommunicationsException, DeploymentException,
        MarshallingException {
        ProgressTracker tracker = progressTrackers.get(jobIdentifier);
        String deploymentStream = tracker.getDeploymentStream(taskIdentifier);

        /*
         * DeploymentMessage deploymentRequest = exchangeFactory
         * .createDeploymentStatusOrAbortRequest(jobIdentifier, taskIdentifier,
         * jobberStream, DeploymentMessage.STATUS);
         */
        ExecutionStatus executionStatus =
            exchangeFactory.createExecutionStatusRequest(jobIdentifier,
                                                         taskIdentifier,
                                                         jobberStream);
        sendExchangeRequest(executionStatus, deploymentStream);

    }


    /**
     * This method returns a String representation of the status of all jobs
     * that have been submitted from this node.
     * 
     * @return A String representation of the status of all jobs.
     */
    public String getStatus() {
        StringBuffer strBuf = new StringBuffer();
        strBuf.append("Number of jobs =" + progressTrackers.size());

        Enumeration<String> jobIds = progressTrackers.keys();
        if (jobIds == null) {
            System.out.println(moduleName + "No jobs have been submitted yet!");
            return strBuf.toString();
        }

        while (jobIds.hasMoreElements()) {
            String jobId = jobIds.nextElement();
            ProgressTracker progressTracker = getStatusOfJob(jobId);
            strBuf.append(progressTracker.toString());
        }

        strBuf.append("Status Report complete!");
        return strBuf.toString();
    }


    /**
     * Retrieve the status of a specific job that was submitted from this node.
     * 
     * @param jobIdentifier
     *            The id of the job in question
     * @return <code>ProgressTracker</code> associated with the job, or NULL if
     *         the specified id is an invalid one.
     */
    public ProgressTracker getStatusOfJob(String jobIdentifier) {
        return progressTrackers.get(jobIdentifier);
    }


    /**
     * Get the complete list of jobs that have been submitted from this node.
     * 
     * @return An enumeration of the submitted jobs.
     */
    public Enumeration<String> getListOfJobs() {
        return progressTrackers.keys();
    }


    @Override
    public synchronized void onEvent(NBEvent nbEvent) {
        byte[] exchangeBytes = nbEvent.getContentPayload();
        int exchangeType = exchangeFactory.getExchangeType(exchangeBytes);

        try {
            if (exchangeType == ExchangeType.DEPLOYMENT) {

                DeploymentMessage deploymentResponse =
                    exchangeFactory.getDeploymentMessage(exchangeBytes);
                String jobIdentifier = deploymentResponse.getJobIdentifier();
                String taskIdentifier = deploymentResponse.getTaskIdentifier();

                int exchangeSubType = deploymentResponse.getExchangeSubType();
                int conditionFlag = deploymentResponse.getConditionFlag();

                if (exchangeSubType == DeploymentMessage.DEPLOY) {
                    if (conditionFlag == ConditionFlag.SUCCESS) {
                        // System.out.println(deploymentResponse);
                    }

                }
                /** End DeploymentMessage.DEPLOY */
            }

            if (exchangeType == ExchangeType.EXECUTION_STATUS) {
                ExecutionStatus executionStatus =
                    exchangeFactory.getExecutionStatus(exchangeBytes);
                TaskLifecycleMetrics taskLifecycleMetrics =
                    executionStatus.getTaskLifecycleMetrics();
                String jobIdentifier = executionStatus.getJobIdentifier();
                String taskIdentifier = executionStatus.getTaskIdentifier();
                String respondTo = executionStatus.getRespondTo();

                ProgressTrackerWidget tracker =
                    progressTrackers.get(jobIdentifier);
                tracker.updateTaskExecutionStatus(trackerBinding,
                                                  taskIdentifier,
                                                  taskLifecycleMetrics,
                                                  respondTo);
                // System.out.println(executionStatus);
            }

            if (exchangeType == ExchangeType.RESOURCE_DISCOVERY) {
                ResourceDiscovery resourceDiscoveryResponse =
                    exchangeFactory.getResourceDiscovery(exchangeBytes);
                processResourceDiscoveryResponse(resourceDiscoveryResponse);
            }

        } catch (IOException e) {
            e.printStackTrace();
        } catch (MarshallingException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }


    private void processResourceDiscoveryResponse(
                                                  ResourceDiscovery resourceDiscoveryResponse) {
        int conditionFlag = resourceDiscoveryResponse.getConditionFlag();
        if (conditionFlag == ConditionFlag.PROBLEMS_ENCOUNTERED) {
            System.out.println(moduleName
                + resourceDiscoveryResponse.getDescription());
            return;
        }

        ResourceMetrics resourceMetrics =
            resourceDiscoveryResponse.getResourceMetrics();
        if (resourceMetrics == null) {
            String report =
                moduleName + "Problems encountered with the "
                    + "discovery response -> "
                    + resourceDiscoveryResponse.getDescription() + " FROM "
                    + resourceDiscoveryResponse.getRespondTo();
            System.out.println(report);
            return;
        }

        String deploymentStream = resourceMetrics.getDeploymentStream();
        if (!resources.containsKey(deploymentStream)) {
            resources.put(deploymentStream, resourceMetrics);
            resourceOrder.add(deploymentStream);
            // System.out.println(moduleName + "Discovered Resource ->\n"
            // + resourceMetrics);
        }

    }


    public ProgressTracker deploy(Operation[] operations, int numOfResources,
                                  String jobIdentifier,
                                  JobLifecycleObserver jobLifecycleObserver)
        throws CommunicationsException, DeploymentException,
        MarshallingException {

        int numOfTasks = operations.length;
        int availableResources = resources.size();

        /* Check to see if the operations or any of the entries is NULL. */
        if (operations == null) {
            throw new DeploymentException("Computations being deployed is NULL");
        }

        for (int i = 0; i < numOfTasks; i++) {
            if (operations[i] == null) {
                throw new DeploymentException("Element (" + i
                    + ") in the set of computations being deployed is NULL.");
            }
        }
        /* Check on the operations is now complete. */

        if (availableResources < numOfResources) {
            String errorReport =
                moduleName + " Requested resources=" + numOfResources
                    + ", exceeds available resources =" + availableResources;
            throw new DeploymentException(errorReport);
        }

        for (int i = 0; i < numOfTasks;) {
            for (int j = 0; j < numOfResources; j++) {
                // int index = j + 1;
                String deployStreamForResource = resourceOrder.elementAt(j);
                submit(operations[i], deployStreamForResource, jobIdentifier,
                       jobLifecycleObserver);
                if (debug) {
                    System.out.println("Deployed operation[" + i
                        + "] on deployment stream: " + deployStreamForResource);
                }
                i++;
                if (i == numOfTasks) {
                    break;
                }
            }
        }

        ProgressTracker tracker = progressTrackers.get(jobIdentifier);
        return tracker;

    }


    public int getNumberOfResources() {
        return resources.size();
    }


    private void sendExchangeRequest(Exchange exchangeRequest,
                                     String resourceDeploymentStream)
        throws CommunicationsException, MarshallingException {
        String errorReport = null;
        try {
            byte[] exchangeRequestBytes = exchangeRequest.getBytes();
            publishToStream(resourceDeploymentStream, exchangeRequestBytes);
            if (debug) {
                System.out.println("Sent deployment request " + exchangeRequest);
            }
        } catch (IOException e) {
            errorReport =
                "Problems marshalling the exchange: " + e.getMessage();
            throwMarshallingException(errorReport);
        }
    }


    /**
     * Utility method which allows a consumer to consume a specific stream
     * 
     * @param synopsisType
     *            The synopsis type of the underlying stream
     * @param streamSynopsis
     *            The synopsis for the stream
     * @throws CommunicationsException
     *             When communication problems are encountered.
     */
    private void consumeStream(String streamSynopsis, int synopsisType)
        throws CommunicationsException {
        try {
            Profile profile =
                streamingService.createProfile(streamSynopsis, synopsisType);
            consumer.subscribeTo(profile);
        } catch (ServiceException serEx) {
            throw new CommunicationsException(serEx.toString());
        }
    }


    /**
     * Utility method that allows a producer to publish a set of bytes onto the
     * specified stream.
     * 
     * @param publishStream
     *            Synopsis of the stream to publish to
     * @param transferBytes
     *            the payload for the stream fragments
     * @throws CommunicationsException
     *             When communication problems are encountered.
     */

    private void publishToStream(String publishStream, byte[] transferBytes)
        throws CommunicationsException {
        NBEvent nbEvent;
        try {
            nbEvent =
                producer.generateEvent(TemplateProfileAndSynopsisTypes.STRING,
                                       publishStream, transferBytes);
            producer.publishEvent(nbEvent);
        } catch (ServiceException e) {
            throwCommunicationsException(e.toString());
        }
    }


    /**
     * Throws a MarshallingException with the specified reason
     * 
     * @param errorReport
     * @throws MarshallingException
     */
    private void throwMarshallingException(String errorReport)
        throws MarshallingException {
        throw new MarshallingException(errorReport);
    }


    /**
     * Throws a Deployment Exception with the specified reason
     * 
     * @param reason
     * @throws DeploymentException
     */
    private void throwDeploymentException(String reason)
        throws DeploymentException {
        throw new DeploymentException(reason);
    }


    /**
     * Throws a CommunicationsException with the specified reason
     * 
     * @param reason
     * @throws CommunicationsException
     */
    private void throwCommunicationsException(String reason)
        throws CommunicationsException {
        throw new CommunicationsException(reason);
    }

}
